package com.cl.ks.flow.handler;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.ks.entity.ClueAutoResult;
import com.cl.ks.entity.KsFilterStatistics;
import com.cl.ks.entity.KsRule;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.flow.enums.NodeHandlerEnum;
import com.cl.ks.flow.util.RuleUtil;
import com.cl.ks.flow.util.StaticCollection;
import com.cl.ks.service.ClueAutoResultService;
import com.cl.ks.service.KsFilterStatisticsService;
import com.cl.mvc.service.BaseService;
import com.cl.spark.dto.SparkResult;
import com.querydsl.jpa.impl.JPAQueryFactory;
import lombok.extern.log4j.Log4j;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;

import static com.cl.ks.constant.QTableConstant.*;

@Slf4j
@Component
public class PushClueHandler extends BaseFlowNodeHandler {

    @Autowired
    private ClueAutoResultService clueAutoResultService;
    @Autowired
    private JPAQueryFactory queryFactory;
    @Autowired
    private RuleUtil ruleUtil;
    @Autowired
    private KsFilterStatisticsService ksFilterStatisticsService;

    @Override
    public SparkResult process(ProcessParam processParam) {
        JSONArray res = new JSONArray();

        JSONObject expression = processParam.getSparkNode().getNodeExpression();
        String sparkNodeCode = processParam.getSparkNode().getCode();
        JSONArray rFields = (JSONArray) expression.get("rFields");
        JSONArray jFields = (JSONArray) expression.get("jFields");
        JSONArray bFields = (JSONArray) expression.get("bFields");
        String ksFlowGraphPower = processParam.getKsFlowGraph().getResultPower() + "," + processParam.getKsFlowGraph().getPower();

        SparkResult originalSparkResult = processParam.getSparkNode().getInputSparkResultList().values().stream().findFirst().get();
        Dataset<Row> dataset = originalSparkResult.getDataset();
        String filterName = processParam.getKsFlowGraph().getName();
        Integer ksFlowGraphId = processParam.getKsFlowGraph().getId();

        List<String> fieldList = Arrays.asList(dataset.schema().fieldNames());
        log.info("###fieldList:" + fieldList);
//        if (!fieldList.contains("id")) {
//            dataset = dataset.withColumn("id", functions.lit(""));
//        }

        if (!fieldList.contains("msg")) {
            dataset = dataset.withColumn("msg", functions.lit(""));
        }

        StaticCollection.newClueAutoResultList(ksFlowGraphId);
        StaticCollection.newRuleIdListMap(ksFlowGraphId);

        dataset.foreach(row -> {
            StringBuilder powerBuilder = new StringBuilder(",");
            if (rFields != null && !rFields.isEmpty()) {
                for (int i = 0; i < rFields.size(); i++) {
                    String s = rFields.getString(i);
                    if (!Objects.isNull(row.getAs(s))) {
                        powerBuilder.append("r_").append(row.getAs(s).toString()).append(",");
                    }
                }
            }
            if (jFields != null && !jFields.isEmpty()) {
                for (int i = 0; i < jFields.size(); i++) {
                    String s = jFields.getString(i);
                    if (!Objects.isNull(row.getAs(s))) {
                        powerBuilder.append("j_").append(row.getAs(s).toString()).append(",");
                    }
                }
            }
            if (bFields != null && !bFields.isEmpty()) {
                for (int i = 0; i < bFields.size(); i++) {
                    String s = bFields.getString(i);
                    if (!Objects.isNull(row.getAs(s))) {
                        powerBuilder.append("b_").append(row.getAs(s).toString()).append(",");
                    }
                }
            }
            powerBuilder.append(",").append(ksFlowGraphPower);
            String power = powerBuilder.toString();
            String ruleName = null;
            String ruleIdStr = null;
            Integer ruleId = null;
            try {
                ruleName = row.getAs("ruleName");
                ruleIdStr = row.getAs("ruleId");
                ruleId = Integer.parseInt(ruleIdStr);
            } catch (Exception e) {
                log.error(filterName + "缺少规则");
            }

            ClueAutoResult clueAutoResult = new ClueAutoResult();
            clueAutoResult.setRuleId(ruleId);
            clueAutoResult.setRuleName(ruleName);
            clueAutoResult.setFilterId(ksFlowGraphId);
            clueAutoResult.setFilterName(filterName);
            clueAutoResult.setDbCode(row.getAs("DB_CODE"));
            clueAutoResult.setTableName(row.getAs("TABLE_NAME"));
            String sourceId = null;
            try {
                sourceId = row.getAs("id").toString();
            } catch (Exception e) {
                log.error(e.getMessage());
                try {
                    sourceId = row.getAs("ID").toString();
                } catch (Exception ee) {
                    log.error(ee.getMessage());
                }
            }
            if (StringUtils.isBlank(sourceId)) {
                throw new RuntimeException("id不存在");
            }
            clueAutoResult.setSourceId(sourceId);
            String msg = "";
            if (!Objects.isNull(row.getAs("msg"))) {
                msg = row.getAs("msg").toString();
            }
            clueAutoResult.setReason(msg);
            clueAutoResult.setTips(msg);
            String powerStr = BaseService.cleanData(RuleUtil.powerConvert(power, row));
            clueAutoResult.setPower(powerStr);
            clueAutoResult.setResultPower(powerStr);
            StaticCollection.getClueAutoResultList(ksFlowGraphId).add(clueAutoResult);
            StaticCollection.getRuleIdListMap(ksFlowGraphId).add(ruleId);
        });

        StaticCollection.getRuleIdListMap(ksFlowGraphId).forEach(ruleId -> {
            if (ruleId != null) {
                KsFilterStatistics ksFilterStatistics = queryFactory.selectFrom(qKsFilterStatistics).where(qKsFilterStatistics.ruleId.eq(ruleId)).orderBy(qKsFilterStatistics.id.desc()).fetchFirst();
                ksFilterStatistics.setClueResultTotal(ksFilterStatistics.getClueResultTotal() + StaticCollection.getClueAutoResultList(ksFlowGraphId).size());
                ksFilterStatisticsService.save(ksFilterStatistics);
            }
        });
        clueAutoResultService.saveAllPartition(StaticCollection.getClueAutoResultList(ksFlowGraphId), 1000, processParam.getKsFlowGraph().getAuth());

        StaticCollection.removeClueAutoResultList(ksFlowGraphId);
        StaticCollection.removeRuleIdListMap(ksFlowGraphId);
        return SparkResult.success(dataset);
    }

    @Override
    public NodeHandlerEnum getType() {
        return NodeHandlerEnum.PUSH_CLUE;
    }
}
